package com.atguigu.flink.sqlfunction;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.TableAggregateFunction;
import org.apache.flink.table.planner.expressions.Collect;
import org.apache.flink.util.Collector;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;

/**
 * Created by Smexy on 2023/2/6
 *
 *  UDTAF:  本质是聚合，输出时输出多行。
 *
 *  示例： 输出每种传感器的前两名的VC
 */
public class Demo5_TableAggFunction
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        String createTableSql = "create table t1( id string,ts bigint ,vc int ) with(" +
            "                    'connector' = 'filesystem' ," +
            "                    'path' = 'data/func/sensor.txt' , " +
            "                    'format' = 'csv'  " +
            "                    )";

        //执行建表(写操作)
        tableEnvironment.executeSql(createTableSql);

        //把t1变为Table对象
        Table table = tableEnvironment.from("t1");

        //1.先创建一个函数对象
        MyUDTAF myFunc = new MyUDTAF();

        //注册函数，再用
        tableEnvironment.createTemporaryFunction("a",myFunc);

        //SQL中目前不支持
        //仅仅支持Tableapi
        table.groupBy($("id"))
             .flatAggregate(call("a",$("vc")))
             //取函数输出的结果的属性名作为列名
             .select($("id"),$("rank"),$("vcNum"))
             .execute()
             .print();
    }

    /*
        必须实现:
            createAccumulator()
            accumulate()
     */
    public static class MyUDTAF extends TableAggregateFunction<MyOutPut,MyAcc>{

        @Override
        public MyAcc createAccumulator() {
            return new MyAcc(0,0);
        }

        //更新累加器的前两名的vc
        public void accumulate(MyAcc myAcc,Integer vc){
            if (vc >= myAcc.firstVc){
                myAcc.secondVc = myAcc.firstVc;
                myAcc.firstVc = vc;
            }else if (vc >= myAcc.secondVc){
                myAcc.secondVc = vc;
            }
        }

        //输出
        public void emitValue(MyAcc acc, Collector<MyOutPut> out){
            //UDTF 来一条数据，输出N行
            out.collect(new MyOutPut(1,acc.firstVc));

            //如果累加器中 secondVc > 0,再输出第二名
            if (acc.secondVc > 0){
                out.collect(new MyOutPut(2,acc.secondVc));
            }
        }
    }

    //输出每种传感器的前两名的VC
    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class MyAcc{
        private Integer firstVc;
        private Integer secondVc;
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class MyOutPut{
        private Integer rank;
        private Integer vcNum;
    }
}
